package com.wzy.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * @Author: wzy
 * @Date: 2022/03/27/16:41
 * @Description:
 */
public class MyConsumerJava {
    private final static String TOPIC_NAME = "my-replicated-topics";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.188.136:9092,192.168.188.136:9093,192.168.188.136:9094");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        //1.创建⼀个消费者的客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String,
                String>(props);
        //2. 消费者订阅主题列表 不同配置如下
        // 指定主题消费
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 指定分区parition 消费
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        // 从头消费
        consumer.beginningOffsets(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        // 指定偏移量消费
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);
        // 指定时间消费
        List<PartitionInfo> topicPartitions =
                consumer.partitionsFor(TOPIC_NAME);
        //从1⼩时前开始消费
        long fetchDataTime = System.currentTimeMillis() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()),
                    fetchDataTime);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap =
                consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry :
                parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() +
                    "|offset-" + offset);
            System.out.println();
            //根据消费⾥的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }

        while (true) {
            /*
             * 3.poll() API 是拉取消息的⻓轮询
             */
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                //4.打印消息
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s %n ", record.partition(),
                        record.offset(), record.key(), record.value());
            }
        }
    }
}
